/**
 * File: BlockingQueueShutdown.java
 * Created by: mhaimel
 * Created on: May 4, 2010
 * CVS:  $Id: BlockingQueueShutdown.java 1.0 May 4, 2010 1:41:26 PM mhaimel Exp $
 */
package uk.ac.ebi.curtain.utils.concurrent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.lang.ObjectUtils.Null;

/**
 * <p>
 * Checks alive status for queue, and returns null
 * </p>
 * <p>
 * Usage: 
 * <code>
 * while(true){
 *   String val = queue.take(); // queue instance of BlockingQueueShutdown<String>
 *   if(null == val && queue.isShutdown()){
 *     // leave loop
 *      break; 
 *   }
 * } // while
 * 
 * </code>
 * </p>
 * 
 * @author mhaimel
 *
 */
public class BlockingQueueShutdown<E> {
	
	private final BlockingQueue<E> queue;
	private final AtomicBoolean shutdown = new AtomicBoolean(false);
	

    /** Lock held by take, put, etc */
    private final ReentrantLock takeLock = new ReentrantLock();

    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();
	
    /**
     * Default queue: {@link LinkedBlockingQueue}
     */
	public BlockingQueueShutdown() {
		this(new LinkedBlockingQueue<E>());
	}	
	
	/**
	 * @param queue {@link BlockingQueue} to be used in this wrapper
	 */
	public BlockingQueueShutdown(BlockingQueue<E> queue) {
		this.queue = queue;
	}
	
	public boolean isShutdown() {
		return this.shutdown.get();
	}
	
	public void shutdown(){
		if(!this.shutdown.getAndSet(true)){ // only signal change once
			signalEmptyStateChange();
		}
	}
	

    /**
     * Signals a waiting take. Called from put / setAlive (which do not
     * otherwise ordinarily lock takeLock.)
     */
    private void signalEmptyStateChange() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
        	if(this.shutdown.get()){
        		notEmpty.signalAll(); // if shutting down - wake up everyone
        	} else {
        		notEmpty.signal();
        	}
        } finally {
            takeLock.unlock();
        }
    }
	
    /**
     * Once {@link #shutdown()} was called, the queue
     * returns {@link Null} values, if the queue is empty 
     * 
     * @return Object NULL or next object in queue  
     * @see {@link BlockingQueue#take(Object)}
     */
    public E take() throws InterruptedException {
        E x = null;
        
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            try {
//                while (count.get() == 0)
            	while(!isShutdown() && this.queue.isEmpty())
                    notEmpty.await();
            } catch (InterruptedException ie) {
                notEmpty.signal(); // propagate to a non-interrupted thread
                throw ie;
            }

            x = extract();
            
        } finally {
            takeLock.unlock();
        }
        return x;
    }

	private E extract() {
		return this.queue.poll();
	}
	
   /**
     * @see {@link BlockingQueue#put(Object)}
     */
    public void put(E e) throws InterruptedException {
    	this.queue.put(e); 
    	signalEmptyStateChange();
    }
    
    public static void main(String[] args) {
		final BlockingQueueShutdown<String> queue = new BlockingQueueShutdown<String>();
		final AtomicBoolean doRun = new AtomicBoolean(true);
		Runnable r = new Runnable() {
			
			@Override
			public void run() {
				try {
					while(doRun.get()){
						String name = queue.take();
						TimeUnit.SECONDS.sleep(1);
						if(null == name){
							if(queue.isShutdown()){
					        	System.out.println(Thread.currentThread().getName() + ": Finshed");
								break;
							}							
						}
					}
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		};
		ExecutorService pool = Executors.newFixedThreadPool(3);
		pool.execute(r);
		pool.execute(r);
		pool.execute(r);
		pool.shutdown();
		
		try {
			for(int i = 0; i < 10; ++i){
				queue.put("Test-"+i);
			}
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		queue.shutdown();
//		doRun.set(false);		
	}

	
}
